package org.apache.hadoop.hdfs.server.blockmanagement;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;

import java.util.Map;

import org.apache.hadoop.hdfs.protocol.Block;
import org.apache.hadoop.hdfs.util.LightWeightLinkedSet;
import org.apache.hadoop.hdfs.server.namenode.NameNode;

/**
 * UnderReplicatedBlocks类用于保存所有等待复制的数据块副本集合。
 * 它维护了一个优先级队列priorityQueues。priorityQueues是一个列表，它有5个子队列，每一个队列对应一个优先级列表，其中0为最高优先级，5为最低优先级。
 */
class UnderReplicatedBlocks implements Iterable<Block> {
  /** The total number of queues : {@value} */
  static final int LEVEL = 5;
  /**
   * 优先级0：保存需要立刻备份的数据块。 这个数据块只有一个副本， 或者这个数据块没有活着的副本并且有一个拷贝所在的Datanode正处于离线中。 这种类型的数据块很有可能丢失
   */
  static final int QUEUE_HIGHEST_PRIORITY = 0;
  /**
   * 优先级1：保存副本数极低的数据块， 当实际副本数与期望副本数的比例小于1:3时， 加入这个队列。
   */
  static final int QUEUE_VERY_UNDER_REPLICATED = 1;
  /**
   * 优先级2：保存正处于备份中的数据块， 但该数据块的副本数并未达到优先级1队列中的比例
   */
  static final int QUEUE_UNDER_REPLICATED = 2;
  /**
   * 优先级3：数据块的副本数量已经足够， 但是数据块副本的分布不是很好， 如果一个机架或者交换机宕机很有可能造成数据块完全丢失。
   */
  static final int QUEUE_REPLICAS_BADLY_DISTRIBUTED = 3;
  /**
   * 优先级4：保存已经损坏的数据块， 就是该数据块对应的所有副本都损坏了。 这里的策略是将损坏的数据块放入这个队列中， 对没有损坏的副本给予更高的优先级
   */
  static final int QUEUE_WITH_CORRUPT_BLOCKS = 4;
  /**
   * 保存所有等待复制的数据块副本集合.它维护了一个优先级队列priorityQueues。
   * priorityQueues是一个列表， 它有5个子队列， 每一个队列对应一个优先级列表， 其中0为最高优先级， 5为最低优先级
   */
  private final List<LightWeightLinkedSet<Block>> priorityQueues = new ArrayList<LightWeightLinkedSet<Block>>();

  /** Stores the replication index for each priority */
  private Map<Integer, Integer> priorityToReplIdx = new HashMap<Integer, Integer>(LEVEL);
  
  /** Create an object. */
  UnderReplicatedBlocks() {
    for (int i = 0; i < LEVEL; i++) {
      priorityQueues.add(new LightWeightLinkedSet<Block>());
      priorityToReplIdx.put(i, 0);
    }
  }

  /**
   * Empty the queues.
   */
  void clear() {
    for (int i = 0; i < LEVEL; i++) {
      priorityQueues.get(i).clear();
    }
  }

  /** Return the total number of under replication blocks */
  synchronized int size() {
    int size = 0;
    for (int i = 0; i < LEVEL; i++) {
      size += priorityQueues.get(i).size();
    }
    return size;
  }

  /** Return the number of under replication blocks excluding corrupt blocks */
  synchronized int getUnderReplicatedBlockCount() {
    int size = 0;
    for (int i = 0; i < LEVEL; i++) {
      if (i != QUEUE_WITH_CORRUPT_BLOCKS) {
        size += priorityQueues.get(i).size();
      }
    }
    return size;
  }

  /** Return the number of corrupt blocks */
  synchronized int getCorruptBlockSize() {
    return priorityQueues.get(QUEUE_WITH_CORRUPT_BLOCKS).size();
  }

  /** Check if a block is in the neededReplication queue */
  synchronized boolean contains(Block block) {
    for(LightWeightLinkedSet<Block> set : priorityQueues) {
      if (set.contains(block)) {
        return true;
      }
    }
    return false;
  }

  /**
   * 用于获取一个指定数据块在UnderReplicatedBlocks集合中的优先级
   */
  private int getPriority(Block block, int curReplicas,  int decommissionedReplicas, int expectedReplicas) {
    assert curReplicas >= 0 : "Negative replicas!";

    if (curReplicas >= expectedReplicas) {
      //数据块的副本数量足够， 但是没有均匀地分布在不同的机架上， 返回优先级3
      return QUEUE_REPLICAS_BADLY_DISTRIBUTED;
    }
    else if (curReplicas == 0) {
      // 当前数据块没有有效的副本， 并且有些副本所在的Datanode正处于离线中， 优先级 0
      if (decommissionedReplicas > 0) {
        return QUEUE_HIGHEST_PRIORITY;
      }
      // 当前数据块的所有副本都是损坏的， 优先级4
      return QUEUE_WITH_CORRUPT_BLOCKS;
    }
    else if (curReplicas == 1) {
      // 只有一个有效副本， 数据块可能随时丢失， 优先级0
      return QUEUE_HIGHEST_PRIORITY;
    }
    else if ((curReplicas * 3) < expectedReplicas) {
      // 当前副本数小于期望副本数的三分之一， 优先级1
      return QUEUE_VERY_UNDER_REPLICATED;
    }
    else {
      // 正常备份状态， 优先级2
      return QUEUE_UNDER_REPLICATED;
    }
  }

  /** add a block to a under replication queue according to its priority
   * @param block a under replication block
   * @param curReplicas current number of replicas of the block
   * @param decomissionedReplicas the number of decommissioned replicas
   * @param expectedReplicas expected number of replicas of the block
   * @return true if the block was added to a queue.
   */
  synchronized boolean add(Block block,
                           int curReplicas, 
                           int decomissionedReplicas,
                           int expectedReplicas) {
    assert curReplicas >= 0 : "Negative replicas!";
    int priLevel = getPriority(block, curReplicas, decomissionedReplicas,
                               expectedReplicas);
    if(priLevel != LEVEL && priorityQueues.get(priLevel).add(block)) {
      if(NameNode.blockStateChangeLog.isDebugEnabled()) {
        NameNode.blockStateChangeLog.debug(
          "BLOCK* NameSystem.UnderReplicationBlock.add:"
          + block
          + " has only " + curReplicas
          + " replicas and need " + expectedReplicas
          + " replicas so is added to neededReplications"
          + " at priority level " + priLevel);
      }
      return true;
    }
    return false;
  }

  /** remove a block from a under replication queue */
  synchronized boolean remove(Block block, 
                              int oldReplicas, 
                              int decommissionedReplicas,
                              int oldExpectedReplicas) {
    int priLevel = getPriority(block, oldReplicas, 
                               decommissionedReplicas,
                               oldExpectedReplicas);
    return remove(block, priLevel);
  }

  /**
   * Remove a block from the under replication queues.
   *
   * The priLevel parameter is a hint of which queue to query
   * first: if negative or &gt;= {@link #LEVEL} this shortcutting
   * is not attmpted.
   *
   * If the block is not found in the nominated queue, an attempt is made to
   * remove it from all queues.
   *
   * <i>Warning:</i> This is not a synchronized method.
   * @param block block to remove
   * @param priLevel expected privilege level
   * @return true if the block was found and removed from one of the priority queues
   */
  boolean remove(Block block, int priLevel) {
    // 从指定队列中删除数据块
    if(priLevel >= 0 && priLevel < LEVEL  && priorityQueues.get(priLevel).remove(block)) {
      return true;
    } else {
      for (int i = 0; i < LEVEL; i++) {
        // 遍历所有队列，查找删除数据块
        if (priorityQueues.get(i).remove(block)) {
          return true;
        }
      }
    }
    return false;
  }

  /**
   * Recalculate and potentially update the priority level of a block.
   *
   * If the block priority has changed from before an attempt is made to
   * remove it from the block queue. Regardless of whether or not the block
   * is in the block queue of (recalculate) priority, an attempt is made
   * to add it to that queue. This ensures that the block will be
   * in its expected priority queue (and only that queue) by the end of the
   * method call.
   * @param block a under replicated block
   * @param curReplicas current number of replicas of the block
   * @param decommissionedReplicas  the number of decommissioned replicas
   * @param curExpectedReplicas expected number of replicas of the block
   * @param curReplicasDelta the change in the replicate count from before
   * @param expectedReplicasDelta the change in the expected replica count from before
   */
  synchronized void update(Block block, int curReplicas,
                           int decommissionedReplicas,
                           int curExpectedReplicas,
                           int curReplicasDelta, int expectedReplicasDelta) {
    int oldReplicas = curReplicas-curReplicasDelta;
    int oldExpectedReplicas = curExpectedReplicas-expectedReplicasDelta;
    int curPri = getPriority(block, curReplicas, decommissionedReplicas, curExpectedReplicas);
    int oldPri = getPriority(block, oldReplicas, decommissionedReplicas, oldExpectedReplicas);
    if(NameNode.stateChangeLog.isDebugEnabled()) {
      NameNode.stateChangeLog.debug("UnderReplicationBlocks.update " + 
        block +
        " curReplicas " + curReplicas +
        " curExpectedReplicas " + curExpectedReplicas +
        " oldReplicas " + oldReplicas +
        " oldExpectedReplicas  " + oldExpectedReplicas +
        " curPri  " + curPri +
        " oldPri  " + oldPri);
    }
    if(oldPri != LEVEL && oldPri != curPri) {
      remove(block, oldPri);
    }
    if(curPri != LEVEL && priorityQueues.get(curPri).add(block)) {
      if(NameNode.blockStateChangeLog.isDebugEnabled()) {
        NameNode.blockStateChangeLog.debug(
          "BLOCK* NameSystem.UnderReplicationBlock.update:"
          + block
          + " has only "+ curReplicas
          + " replicas and needs " + curExpectedReplicas
          + " replicas so is added to neededReplications"
          + " at priority level " + curPri);
      }
    }
  }
  
  /**
   * chooseUnderReplicatedBlocks()方法每次从UnderReplicatedBlocks的优先级队列中选出blocksToProcess个待复制副本，
   * 然后在UnderReplicatedBlocks.priorityToReplIdx字段中记录每个优先级队列的读取偏移量，
   * chooseUnderReplicatedBlocks()方法会在下次调用时从每个优先级队列的读取偏移量位置读取副本。
   * 如果最后一个优先级队列的读取偏移量到达了队列的末尾，也就是最近没有新添加的待复制数据块时，则重置所有优先级队列的读取偏移量为0，然后从优先级最高的优先级队列开始读取操作。
   */
  public synchronized List<List<Block>> chooseUnderReplicatedBlocks( int blocksToProcess) {
    // 初始化返回值列表，保存从每个优先级队列中取出的数据块
    List<List<Block>> blocksToReplicate = new ArrayList<List<Block>>(LEVEL);
    for (int i = 0; i < LEVEL; i++) {
      blocksToReplicate.add(new ArrayList<Block>());
    }

    if (size() == 0) { // underReplicatedBlocks 没有保存任何待复制数据块
      return blocksToReplicate;
    }
    
    int blockCount = 0;
    // 遍历 underReplicatedBlocks 中所有优先级队列
    for (int priority = 0; priority < LEVEL; priority++) { 
      // 当前优先级队列保存的待复制数据块的迭代器
      BlockIterator neededReplicationsIterator = iterator(priority);
      // 获取当前优先级队列的读取偏移量
      Integer replIndex = priorityToReplIdx.get(priority);
      
      // 从 priorityToRepIdx 字段记录的读取游标开始读取数据
      for (int i = 0; i < replIndex && neededReplicationsIterator.hasNext(); i++) {
        neededReplicationsIterator.next();
      }

      // 获取当前队列中读取的副本数量
      blocksToProcess = Math.min(blocksToProcess, size());
      
      if (blockCount == blocksToProcess) {
        break;  // 如果读取了足够的数量，则退出循环
      }
      
      // 读取副本，并将副本保存到 blocksToReplicate 返回值列表中
      while (blockCount < blocksToProcess && neededReplicationsIterator.hasNext()) {
        Block block = neededReplicationsIterator.next();
        blocksToReplicate.get(priority).add(block);
        replIndex++;
        blockCount++;
      }
      
      if (!neededReplicationsIterator.hasNext() && neededReplicationsIterator.getPriority() == LEVEL - 1) {
        // 将所有优先级队列的读取偏移量重置为0，因为最近没有新添加的待复制副本
        for (int i = 0; i < LEVEL; i++) {
          priorityToReplIdx.put(i, 0);
        }
        break;
      }
      // 更新当前队列的读取游标
      priorityToReplIdx.put(priority, replIndex); 
    }
    return blocksToReplicate; // 返回所有待复制数据块
  }

  /** returns an iterator of all blocks in a given priority queue */
  synchronized BlockIterator iterator(int level) {
    return new BlockIterator(level);
  }

  /** return an iterator of all the under replication blocks */
  @Override
  public synchronized BlockIterator iterator() {
    return new BlockIterator();
  }

  /**
   * An iterator over blocks.
   */
  class BlockIterator implements Iterator<Block> {
    private int level;
    private boolean isIteratorForLevel = false;
    private final List<Iterator<Block>> iterators = new ArrayList<Iterator<Block>>();

    /**
     * Construct an iterator over all queues.
     */
    private BlockIterator() {
      level=0;
      for(int i=0; i<LEVEL; i++) {
        iterators.add(priorityQueues.get(i).iterator());
      }
    }

    /**
     * Constrict an iterator for a single queue level
     * @param l the priority level to iterate over
     */
    private BlockIterator(int l) {
      level = l;
      isIteratorForLevel = true;
      iterators.add(priorityQueues.get(level).iterator());
    }

    private void update() {
      if (isIteratorForLevel) {
        return;
      }
      while(level< LEVEL-1 && !iterators.get(level).hasNext()) {
        level++;
      }
    }

    @Override
    public Block next() {
      if (isIteratorForLevel) {
        return iterators.get(0).next();
      }
      update();
      return iterators.get(level).next();
    }

    @Override
    public boolean hasNext() {
      if (isIteratorForLevel) {
        return iterators.get(0).hasNext();
      }
      update();
      return iterators.get(level).hasNext();
    }

    @Override
    public void remove() {
      if (isIteratorForLevel) {
        iterators.get(0).remove();
      } else {
        iterators.get(level).remove();
      }
    }

    int getPriority() {
      return level;
    }
  }

  /**
   * 更新 priority队列的读取偏移量
   */
  public void decrementReplicationIndex(int priority) {
    Integer replIdx = priorityToReplIdx.get(priority);
    priorityToReplIdx.put(priority, --replIdx);  // 将 priority队列的读取偏移量减1
  }
}
